/*  
  -- The Lime II Project -- 

  A tuplespaces-based middleware for coordinating agents and mobile hosts.
  Copyright (C) 2005.
  Gian Pietro Picco, Amy L. Murphy, Lorenzo Bellini.

  This library is free software; you can redistribute it and/or
  modify it under the terms of the GNU Lesser General Public
  License as published by the Free Software Foundation; either
  version 2.1 of the License, or (at your option) any later version.

  This library is distributed in the hope that it will be useful,
  but WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  Lesser General Public License for more details.

  You should have received this copy of the GNU Lesser General Public
  License along with this library; if not, write to the Free Software
  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA 
*/

package lime2;
import lights.interfaces.*;
import java.util.*;

/**
 * Manages ghost tuples. Ghost tuples are tuples whose delivery we are uncertain of; they are temporarily unavailable
 * for (user level) local operations: they are kept in a special inconsistent state which is reconcilied before a 
 * remote standard operation can take place. Although the functionalities og GhostBag could have been provided by 
 * Reconciliator, we choose to use a different class - in order to keep "synchronization" overhead as little as 
 * possible.
 *  
 * @author Gian Pietro Picco
 * @author Amy Murphy
 * @author Lorenzo Bellini
 */

class GhostBag 
{
 // private static data structures	
 private static Hashtable lastTuples  = new Hashtable(); // format: [LimeServerID,Hashtable[name,ITuple]]
 private static Hashtable ghostTuples = new Hashtable(); // format: [LimeServerID,Hashtable[name,Vector]]

 /**
  * Extracts the last tuple carried by this remote op and records it - if any, if it has to be recorded.
  * 
  * @param ro
  */
 static void saveLast(RemoteOp ro)
 {
  ITuple bak = null;	
 	
  // please note that information about the type/subtype of the op (and non null carried tuples) is sufficient to find
  // the right op ("step") within a protocol (.. FailureR forces termination of previous steps)
  
  if ((LimeTSMgr.get(ro.name) != null && LimeTSMgr.get(ro.name).getPolicy() == Policies.CONSERVATIVE) &&	
  (
   (ro.type == LimeConstants.REM_INP_UNI   && ro.subtype == LimeConstants.ACK)          ||  	
   (ro.type == LimeConstants.REM_INP_MULTI && ro.subtype == LimeConstants.ACK)          || 
   (ro.type == LimeConstants.REM_OUT       && ro.subtype == LimeConstants.PASSIVE_OPEN) || 
   (ro.type == LimeConstants.REM_INW       && ro.subtype == LimeConstants.ACK)          ||
   (ro.type == LimeConstants.REM_ING_UNI   && (ro.subtype == LimeConstants.ACK || ro.subtype == LimeConstants.FIN)) ||
   (ro.type == LimeConstants.REM_ING_MULTI && (ro.subtype == LimeConstants.ACK || ro.subtype == LimeConstants.FIN)) ||    
   (ro.type == LimeConstants.REM_OUTG      && (ro.subtype == LimeConstants.ACK || ro.subtype == LimeConstants.FIN))   
   // please note that RECONCILE ops are avoided !
  ))  
  {
   if (ro instanceof RemTupleOp) 	
   { 
    RemTupleOp rto = (RemTupleOp) ro;
    if (rto.tuples != null && rto.tuples.length > 0) bak = rto.tuples[rto.tuples.length - 1];
   }
   else if (ro instanceof RemWeakOp)
   {
    RemWeakOp rwo = (RemWeakOp) ro;
    if (rwo.wevent != null) bak = rwo.wevent.getTuple();
   }
   
   if (bak != null) saveLastTuple(ro.target, ro.name, bak);	
  }
 }
 
 /**
  * Records the last tuple received per host, per tuplespace.
  * 
  * @param target host
  * @param tuplespace name
  * @param last tuple received
  */
 synchronized private static void saveLastTuple(LimeServerID target, String tuplespace, ITuple last)
 {
  Hashtable spaces = (Hashtable) lastTuples.get(target);
  if (spaces == null) { spaces = new Hashtable(); lastTuples.put(target, spaces); }
  spaces.put(tuplespace, last);
 }
 
 /**
  * Returns the last tuple received per host, per tuplespace.
  * 
  * @param target host
  * @param tuplespace name
  * @return last tuple received
  */
 synchronized static ITuple getLastTuple(LimeServerID target, String tuplespace)
 {
  ITuple ret = null;
  Hashtable spaces = (Hashtable) lastTuples.get(target);
  if (spaces != null) ret = (ITuple) spaces.get(tuplespace);
  return ret;
 }
 
 /**
  * Adds ghost tuples per host, per tuplespace; preserves chronological order.
  * 
  * @param target host
  * @param tuplespace name
  * @param tuples ghost tuples
  */
 synchronized static void addGhostTuples(LimeServerID target, String tuplespace, ITuple[] tuples)
 {
  // ghost tuples that come from out/outg operations have a destination location different from this host - we have to
  // deliver them. This (usually) doesn't hold for ghost tuples that come from in/inp/ing operations (if it holds we
  // have to deliver them anyway): ghost tuples that come from in/inp/ing operations must be re-inserted in the local
  // tuplespace if they were not delivered
  if (tuples != null)
  {	
   Hashtable spaces = (Hashtable) ghostTuples.get(target);
   if (spaces == null) { spaces = new Hashtable(); ghostTuples.put(target, spaces); }
   Vector v = (Vector) spaces.get(tuplespace);
   if (v == null) { v = new Vector(); spaces.put(tuplespace, v); }
   for (int i = 0; i < tuples.length; i++) v.addElement(tuples[i]);
  } 
 }
 
 /**
  * Returns ghost tuples (per host, per tuplespace) that are younger than the given (non null) tuple. 
  * 
  * @param target host
  * @param tuplespace name
  * @param young given tuple for comparing "age" 
  * @return array of younger ghost tuples
  */
 synchronized static ITuple[] getYoungerGhostTuples(LimeServerID target, String tuplespace, ITuple young)
 {
  ITuple[] ret = null;
  TupleID age = LimeWrapper.getID(young);
  int start = 0;
  
  Hashtable spaces = (Hashtable) ghostTuples.get(target);
  if (spaces != null)
  {
   Vector v = (Vector) spaces.get(tuplespace);
   if (v != null)
   {
   	int l = v.size(); 
    for (int i = 0; start == 0 && i < l - 1; i++) if (age.equals(LimeWrapper.getID((ITuple) v.get(i)))) start = i + 1;
   	if (l - start > 0)
   	{
   	 ret = new ITuple[l - start];
   	 for (int j = start; j < l; j++) ret[j] = (ITuple) v.get(j);
   	}
   }	
  }
  
  return ret;
 }
 
 /** 
  * Returns all ghost tuples per host, per tuplespace.
  * 
  * @param target host
  * @param tuplespace name
  * @return array of all ghost tuples
  */
 synchronized static ITuple[] getAllGhostTuples(LimeServerID target, String tuplespace)
 {
  ITuple[] ret = null;	
  Hashtable spaces = (Hashtable) ghostTuples.get(target);
  if (spaces != null) 
  { 
   Vector v = (Vector) spaces.get(tuplespace); 
   if (v != null && v.size() > 0) 
   {
    ret = new ITuple[v.size()];
    v.copyInto(ret);
   }	
  }
  return ret;
 }
 
 /**
  * Removes all ghost tuples per host, per tuplespace.
  * 
  * @param target host
  * @param tuplespace name
  */
 synchronized static void removeGhostTuples(LimeServerID target, String tuplespace)
 {
  Hashtable spaces = (Hashtable) ghostTuples.get(target);
  if (spaces != null) spaces.remove(tuplespace);
 }
 
}
